package com.pw.study.dstream


import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.streaming.{StreamingContext, StreamingContextState}

import java.net.URI


class MonitorStop(sc:StreamingContext) extends Runnable{
  override def run(): Unit = {

    val fs: FileSystem = FileSystem.get(new URI("hdfs://hadoop112:8020"),new Configuration(),"atguigu")

    while (true){
      Thread.sleep(5000)
      // 获取/stopSpark路径是否存在
      val result: Boolean = fs.exists(new Path("hdfs://hadoop112:8020/stopSpark"))

      if (result){

        val state: StreamingContextState = sc.getState()
        // 获取当前任务是否正在运行
        if (state == StreamingContextState.ACTIVE){
          // 优雅关闭
          sc.stop(stopSparkContext = true, stopGracefully = true)
          System.exit(0)
        }
      }
    }
  }

}
